package com.flink.window;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;

import java.time.Duration;

/**
 * 描述:
 * 其他api
 *
 * @author yanzhengwu
 * @create 2022-07-31 21:22
 */
public class WindowOtherApi {


        public static void main(String[] args) throws Exception {

            //声明执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
            env.setParallelism(1);
            //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
            env.getConfig().setAutoWatermarkInterval(100);


            //TODO 无序流的watermark生成策略
            DataStream<User> stream = env.addSource(new UserSource())       //生成水位线和时间戳的策略对象
                    .assignTimestampsAndWatermarks(
                            //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                            WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                    //返回策略对象的具体实现
                                    .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                        /**
                                         * 此方法为指定以事件时间的具体时间戳字段
                                         *
                                         * @param element
                                         * @param recordTimestamp
                                         * @return 返回的则是一个毫秒数的时间戳
                                         */
                                        @Override
                                        public long extractTimestamp(User element, long recordTimestamp) {
                                            return element.getTimestamp();
                                        }
                                    }));
            //打印数据和 计算结果进行区分
            stream.print("data==>");
            //数据倾斜就是为了在一定场景下必须等所有数据都到期了在进行计算
            stream.keyBy(data -> true)
                    .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                    .trigger(new MyTrigger())
                    .evictor(new MyEvictor())
                    //TODO 以下方法为允许延迟，由于watermark把时钟调慢这把锁太重导致整个链路都慢了；所以在窗口函数关闭时 还允许迟到（10分钟）数据参与计算
                    //TODO 就相当于公交车发车了 走的很慢而且是开着门走的，所以数据依然可以上车计算
                    .allowedLateness(Time.minutes(10))
                    .sum("")
                    .print();

            env.execute();
        }
        //TODO 触发器  告诉flink什么情况下执行窗口计算
        public static class MyTrigger extends Trigger<User, TimeWindow> {


            @Override
            public TriggerResult onElement(User element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

            }
        }

        //TODO 移除器相当于过滤数据的概念 移除指定那些数据不参与计算
        public static class MyEvictor implements Evictor<User,TimeWindow> {


            @Override
            public void evictBefore(Iterable<TimestampedValue<User>> elements, int size, TimeWindow window, EvictorContext evictorContext) {

            }

            @Override
            public void evictAfter(Iterable<TimestampedValue<User>> elements, int size, TimeWindow window, EvictorContext evictorContext) {

            }
        }
}
